#include <unordered_map>
#include <unordered_set>
#include <cerrno>
#include <cstring>
#include <iostream>
#include <vector>
#include <string>
#include <sys/sendfile.h>
#include <sys/types.h>
#include <sys/stat.h>
#include <signal.h>
#include <unistd.h>
#include "socket.hh"
#include "epoller.hh"
#include "../log/log.hpp"
using std::cerr;
using std::cout;
using std::endl;
using std::string;
using std::to_string;
using std::vector;
/*
** Transport (传输层):
**
**  制定了用于传输报文的策略和结构:
**  包含: epoll多路转接的IO策略;
**        reactor反应堆模式;
**        Tcp 可靠传输协议
*/
class tcp_reactor_t
{
    using service_call_back_t = std::function<int(connection_t &, string &)>;
    using package_split_call_back_t = std::function<bool(string &, vector<string> &)>;
    friend class connection_t;

private:
    // 网络
    tcp_socket_t _socket;
    // 多路转接
    epoller_t _epoller;
    //  系统多路转接epoll api与封装对象的转换
    //  有了这张表,从epoll就绪队列拿到epoll_event.data.fd ->直接找到对应的connection对象进而调用回调
    //  connections都需要在堆上才具有稳定的生命
    std::unordered_map<int, connection_t *> _connect_map;
    std::unordered_set<connection_t *> _exits_set;
    // 业务处理
    service_call_back_t _service_handler;
    package_split_call_back_t _package_spliter;

public:
    tcp_reactor_t(const service_call_back_t &service_func, const package_split_call_back_t &package_split_func, const uint16_t port)
        : _service_handler(service_func), _package_spliter(package_split_func), _socket(port) { ; }
    ~tcp_reactor_t()
    {
        for (connection_t *e : _exits_set)
            del_connection(e);
    }

private:
    void init_reactor()
    {
        // 防止写入时进程崩溃
        signal(SIGPIPE, SIG_IGN);
        // 开始监听
        _socket.listen_fd(128);
        // 设置listen_fd 非阻塞
        _socket.set_sock_nonblock();
        // 添加listen关心
        // 设置listen的读就绪方法 添加进epoll模型并建立映射
        add_connection(_socket.get_fd(), EPOLLIN | EPOLLET, std::bind(&tcp_reactor_t::accepter, this, std::placeholders::_1), nullptr, nullptr);
        log(INFO, "reactor inited");
        log(INFO, "socket: " + to_string(_socket.get_fd()));
    }
    // 获取新连接的方法
    bool accepter(connection_t &connection)
    {
        // 新链接
        // listen 也是et模式 要循环获取新套接字
        while (true)
        {
            int new_fd = _socket.accept_fd();
            if (new_fd < 0)
            {
                if (errno == EAGAIN || errno == EWOULDBLOCK)
                    break;
                else if (errno == EINTR)
                    continue;
                else
                    return false;
            }
            else
            {
                log(INFO, "accepted: " + to_string(new_fd));
                // 设置非阻塞
                if (!set_fd_nonblock(new_fd))
                    log(ERROR, "fd faild to set nonblock");
                // 设置方法 添加新连接
                add_connection(new_fd, EPOLLIN | EPOLLERR | EPOLLRDHUP | EPOLLET,
                               std::bind(&tcp_reactor_t::recv_handler, this, std::placeholders::_1),
                               std::bind(&tcp_reactor_t::send_handler, this, std::placeholders::_1),
                               std::bind(&tcp_reactor_t::error_handler, this, std::placeholders::_1));
            }
        }
        return true;
    }
    bool recv_handler(connection_t &connection)
    {
        int fd = connection.get_fd();
        char temp_buffer[2048];
        while (true)
        {
            bzero(temp_buffer, 2048);
            ssize_t recv_ret = recv(fd, temp_buffer, 2048, MSG_DONTWAIT);
            if (recv_ret > 0)
            {
                // 输入connection缓冲区
                connection._in_buffer.append(temp_buffer, recv_ret);
            }
            else
            {
                if (recv_ret == 0)
                {
                    // 对端关闭就无需在处理报文直接退出
                    log(INFO, "recv_client_quit");
                    connection.error_handler();
                    return true;
                }
                else
                {
                    // 内核缓冲区读完了
                    if (errno == EAGAIN || errno == EWOULDBLOCK)
                        break;
                    // 被信号打断
                    else if (errno == EINTR)
                        continue;
                    else
                    {
                        // 异常处理
                        if (connection.is_func_exist_error())
                            connection.error_handler();
                        return false;
                    }
                }
            }
        }
        // 完成一次et触发的读取: I/O完成
        // 提取报文
        vector<string> split_res;
        // 调用报文分割回调
        if (_package_spliter(connection._in_buffer, split_res) == false)
            if (connection.is_func_exist_error())
            {
                // 异常处理
                connection.error_handler();
                log(INFO, "error handler");
                return false;
            }
        // 上交应用层
        for (auto &e : split_res)
        {
            // cout<<"to service pool:"<<e;
            //  可加入线程池
            //  调用报文处理回调 ->处理业务并把结果写入读缓冲区打开关心
            _service_handler(connection, e);
        }
        return true;
    }
    bool send_handler(connection_t &connection)
    {
        bool buffer_done = connection._out_buffer.empty();
        bool file_done = connection._sendfile_fd < 0;

        while (!(buffer_done && file_done))
        {
            // 发送body正文
            if (!buffer_done)
            {
                // 非阻塞 send
                ssize_t send_ret = send(connection.get_fd(), connection._out_buffer.c_str(), connection._out_buffer.size(), MSG_DONTWAIT);
                if (send_ret > 0)
                {
                    connection._out_buffer.erase(0, send_ret);
                    buffer_done = connection._out_buffer.empty();
                }
                else
                {
                    if (errno == EINTR)
                        continue;
                    if (errno == EWOULDBLOCK || errno == EAGAIN)
                        break;
                    // 对端关闭或其他错误
                    buffer_done = true;
                    // close file fd=-1
                    close_file(connection._sendfile_fd);
                    // 关闭连接
                    log(WARNING, "send content error");
                    connection.error_handler();
                    return false;
                }
            }

            // 调用sendfile发送文件
            if (!file_done)
            {
                ssize_t remain_bytes = connection._file_size;
                // sendfile 可能不会一次发送完毕
                while (remain_bytes > 0)
                {
                    ssize_t once_sent = sendfile(connection.get_fd(), connection._sendfile_fd, nullptr, remain_bytes);
                    if (once_sent <= 0)
                    {
                        if (errno == EINTR)
                            continue;
                        if (errno == EWOULDBLOCK || errno == EAGAIN)
                        {
                            // 保持写事件的打开，更新下次发送的大小
                            connection._file_size = remain_bytes;
                            break;
                        }
                        // 对端关闭或其他错误
                        file_done = true;
                        // close file fd=-1
                        close_file(connection._sendfile_fd);
                        // 关闭连接
                        if (connection.is_func_exist_error())
                        {
                            connection.error_handler();
                            return false;
                        }
                    }
                    else
                    {
                        // 更新下次发送的大小
                        remain_bytes -= once_sent;
                    }
                }
                // 判断发送是否完毕
                if (remain_bytes <= 0)
                {
                    file_done = true;
                    close_file(connection._sendfile_fd);
                }
            }
        }
        // 强调 数据发送完毕才关闭可写事件关心
        if (buffer_done && file_done)
        {
            del_write_care(connection);
        }

        return true;
    }
    bool error_handler(connection_t &connection)
    {
        return del_connection(&connection);
    }

    // 添加链接
    void add_connection(const int fd, uint32_t care, const handler_t &recv_func = handler_t(), const handler_t &send_func = handler_t(), const handler_t &error_func = handler_t())
    {
        // 创建connection对象
        //  设置读写异常方法 以及关心事件
        connection_t *p_connection = new connection_t(fd, this, care, recv_func, send_func, error_func);
        // 将connection添加进epoll模型
        _epoller.add_connection(p_connection);
        // 建立映射
        _connect_map[p_connection->get_fd()] = p_connection;
        // 保存信息
        _exits_set.insert(p_connection);
    }

    // 删除链接
    bool del_connection(connection_t *p_connection)
    {
        bool result = true;
        if (!p_connection || !is_con_exist(p_connection))
        {
            log(ERROR, string("p_con not exists"));
            log(ERROR, string("del_connection failed!"));
            result = false;
        }
        else
        {
            // 从epoll模型中删除
            _epoller.del_connection(p_connection);
            // 删除映射
            _connect_map.erase(p_connection->get_fd());
            // 抹除信息
            _exits_set.erase(p_connection);
            // 关闭网络socket
            close(p_connection->get_fd());
            log(INFO, "closed: " + to_string(p_connection->get_fd()));
            // 释放内存
            delete p_connection;
            p_connection = nullptr;
        }
        return result;
    }
    // 辅助方法 closefd
    bool close_file(int &fd)
    {
        bool ret = close(fd);
        if (!ret)
            log(INFO, "closed fd: " + to_string(fd));
        fd = -1;
        return !ret;
    }
    // 辅助方法 设置非阻塞
    bool set_fd_nonblock(int &fd)
    {
        int falg = fcntl(fd, F_GETFL);
        return fcntl(fd, F_SETFL, falg | O_NONBLOCK) != -1;
    }

public:
    // 查询是否存在连接
    bool is_con_exist(connection_t *p_connection)
    {
        return _exits_set.count(p_connection);
    }
    // 开关写事件关心 一般不常打开
    bool add_write_care(connection_t &connection)
    {
        return _epoller.add_care(connection, EPOLLOUT);
    }
    bool del_write_care(connection_t &connection)
    {
        return _epoller.del_care(connection, EPOLLOUT);
    }
    // 重载
    bool add_write_care(connection_t *p_connection)
    {
        return _epoller.add_care(*p_connection, EPOLLOUT);
    }
    bool del_write_care(connection_t *p_connection)
    {
        return _epoller.del_care(*p_connection, EPOLLOUT);
    }
    void run()
    {
        // 初始化
        init_reactor();
        while (true)
        {
            loop_once();
        }
    }
    bool loop_once()
    {
        // 等
        log(INFO, "waiting...");
        int ready_num = _epoller.wait();
        log(INFO, "get");
        if (ready_num < 0)
        {
            if (errno == EINTR)
                return true;
            log(FATAL, string("epoll_wait error!").append(strerror(errno)));
            return true;
        }
        // 事件派发
        for (int i = 0; i < ready_num; i++)
        {
            dispatch(_epoller.get_event_by_index(i));
        }
        return true;
    }
    void dispatch(const epoll_event *p_evt)
    {
        if (!p_evt)
        {
            log(WARNING, "p_evt is nullptr!");
            return;
        }
        // 查找映射 fd->con
        auto iter = _connect_map.find(p_evt->data.fd);
        if (iter == _connect_map.end())
        {
            log(WARNING, "connection not found");
            return;
        }
        connection_t &con = *iter->second;
        uint32_t events = con.get_events();
        if (events & EPOLLERR || events & EPOLLRDHUP)
        {
            // 统一异常处理
            events |= (EPOLLIN | EPOLLOUT);
        }
        if (events & EPOLLIN)
        {
            // 每一个事件被处理都可能关闭连接 加强判断提高服务器壮性
            if (is_con_exist(&con) && con.is_func_exist_recv())
            {
                con.recv_handler();
            }
        }
        if (events & EPOLLOUT)
        {
            if (is_con_exist(&con) && con.is_func_exist_send())
            {
                con.send_handler();
            }
        }
    }
};
